airbyte_cdk/sources/declarative/transformations/config_transformations/config_transformation.py (1)

9-23: Type hint mismatch between base class and implementation, wdyt?

The transform method's parameter type is Dict[str, Any], but the implementation in remap_field.py uses MutableMapping[str, Any]. Since the method is meant to modify the config in-place, MutableMapping seems more appropriate here. Would you consider changing this to be consistent with the implementation?

-    def transform(
-        self,
-        config: Dict[str, Any],
-    ) -> None:
+    def transform(
+        self,
+        config: MutableMapping[str, Any],
+    ) -> None:

Also, you might want to add the missing import:

-from typing import Any, Dict
+from typing import Any, Dict, MutableMapping

airbyte_cdk/sources/declarative/transformations/config_transformations/remap_field.py (2)

23-34: Possible redundancy in field path conversion, wdyt?

The initialization logic first converts all paths to InterpolatedString using a list comprehension, then loops through them again to check if they're strings and convert them. Could we simplify this logic since we're already handling everything in the list comprehension?

    def __post_init__(self) -> None:
        if not self.field_path:
            raise Exception("field_path cannot be empty.")
-        self._field_path = [
-            InterpolatedString.create(path, parameters={}) for path in self.field_path
-        ]
-        for path_index in range(len(self.field_path)):
-            if isinstance(self.field_path[path_index], str):
-                self._field_path[path_index] = InterpolatedString.create(
-                    self.field_path[path_index], parameters={}
-                )
+        self._field_path = []
+        for path in self.field_path:
+            if isinstance(path, str):
+                self._field_path.append(InterpolatedString.create(path, parameters={}))
+            else:
+                self._field_path.append(path)

Also, consider using a more specific exception type with a clearer error message:

        if not self.field_path:
-            raise Exception("field_path cannot be empty.")
+            raise ValueError("field_path cannot be empty. Please provide at least one path component.")

---

35-60: Implementation seems robust!

The transform logic safely navigates the nested dictionary structure and only applies transformations when appropriate. I like the early returns when paths don't exist or aren't mappings, which makes the code cleaner and prevents errors.

One suggestion - consider adding a log statement when a remapping occurs, which could be helpful for debugging:

        if field_name in current and current[field_name] in self.map:
+            # Could add logging here if needed
+            # e.g., logger.debug(f"Remapping {field_name} from {current[field_name]} to {self.map[current[field_name]]}")
            current[field_name] = self.map[current[field_name]]

airbyte_cdk/sources/declarative/validators/predicate_validator.py (1)

1-27: Simple but effective validator implementation, wdyt about a small enhancement?

The PredicateValidator class looks well-structured with a clear single responsibility - it stores a value and delegates validation to a provided strategy. The implementation is clean and the docstrings are clear.

I notice that this class doesn't inherit from the Validator abstract base class, while it's placed in the validators package. This might be intentional based on the different method signatures (validate() vs validate(input_data: Any)), but it could be worth considering whether it should implement the common interface for consistency. Alternatively, renaming it to distinguish its different role might be helpful.

Would adding type validation for the strategy parameter in the constructor make sense to catch configuration errors earlier? Just a thought!

airbyte_cdk/sources/declarative/validators/validate_adheres_to_schema.py (2)

21-31: Implementation is concise and straightforward

Nice implementation of the schema validation strategy! It's clean and effectively wraps the jsonschema library while providing a good error message.

One question: should we also catch jsonschema.SchemaError separately? It happens when the schema itself is invalid (as shown in your test_given_invalid_schema_when_validate_then_raises_error test), and we might want to distinguish between invalid schema vs invalid data in the error message, wdyt?

try:
    jsonschema.validate(instance=value, schema=self.schema)
except jsonschema.ValidationError as e:
    raise ValueError(f"JSON schema validation error: {e.message}")
+except jsonschema.SchemaError as e:
+    raise ValueError(f"Invalid JSON schema: {e.message}")

---

21-31: Suggestion: Add support for JSON string inputs

I noticed that your test suite includes a test for validating JSON strings (test_given_json_string_when_validate_then_succeeds), but the implementation doesn't explicitly handle JSON string inputs. The test is passing, which suggests jsonschema might handle this automatically, but it could be more robust to add explicit handling, wdyt?

def validate(self, value: Any) -> None:
    """
    Validates the value against the JSON schema.

    :param value: The value to validate
    :raises ValueError: If the value does not adhere to the schema
    """
+    # If value is a JSON string, parse it first
+    import json
+    if isinstance(value, str):
+        try:
+            value = json.loads(value)
+        except json.JSONDecodeError as e:
+            raise ValueError(f"Invalid JSON string: {e}")
+            
    try:
        jsonschema.validate(instance=value, schema=self.schema)
    except jsonschema.ValidationError as e:
        raise ValueError(f"JSON schema validation error: {e.message}")

airbyte_cdk/sources/declarative/validators/dpath_validator.py (3)

25-34: Simplify the path initialization

The current implementation of __post_init__ looks like it's doing the same conversion twice. You first create self._field_path by converting all elements, then you iterate through self.field_path again to convert any string elements. I think we can simplify this, wdyt?

def __post_init__(self) -> None:
-    self._field_path = [
-        InterpolatedString.create(path, parameters={}) for path in self.field_path
-    ]
-    for path_index in range(len(self.field_path)):
-        if isinstance(self.field_path[path_index], str):
-            self._field_path[path_index] = InterpolatedString.create(
-                self.field_path[path_index], parameters={}
-            )
+    self._field_path = [
+        InterpolatedString.create(path, parameters={}) for path in self.field_path
+    ]

---

47-65: Refactor to reduce code duplication

The validation logic for wildcard and non-wildcard paths contains duplicated code. We could extract the common validation logic into a helper method to make the code more DRY, wdyt?

if "*" in path:
    try:
        values = dpath.values(input_data, path)
    except KeyError as e:
        raise KeyError(f"Error validating path '{self.field_path}': {e}")
    for value in values:
-        try:
-            self.strategy.validate(value)
-        except Exception as e:
-            raise ValueError(f"Error validating value '{value}': {e}")
+        self._validate_value(value)
else:
    try:
        value = dpath.get(input_data, path)
    except KeyError as e:
        raise KeyError(f"Error validating path '{self.field_path}': {e}")
-    try:
-        self.strategy.validate(value)
-    except Exception as e:
-        raise ValueError(f"Error validating value '{value}': {e}")
+    self._validate_value(value)

+def _validate_value(self, value: Any) -> None:
+    try:
+        self.strategy.validate(value)
+    except Exception as e:
+        raise ValueError(f"Error validating value '{value}': {e}")

---

35-42: Consider exposing a config parameter for path interpolation

Currently, you're passing an empty dictionary to the eval method. If the field paths contain Jinja templates, they would need context to be properly evaluated. Would it make sense to accept a config parameter in the validate method and pass that to the eval calls, wdyt?

-def validate(self, input_data: dict[str, Any]) -> None:
+def validate(self, input_data: dict[str, Any], config: dict[str, Any] = None) -> None:
    """
    Extracts the value at the specified path and applies the validation strategy.

    :param input_data: Dictionary containing the data to validate
+   :param config: Optional configuration for interpolating path elements
    :raises ValueError: If the path doesn't exist or validation fails
    """
-    path = [path.eval({}) for path in self._field_path]
+    config = config or {}
+    path = [path.eval(config) for path in self._field_path]

unit_tests/sources/declarative/validators/test_predicate_validator.py (1)

1-7: Add missing copyright header

The copyright header is missing from this file. All other files in this PR have it, wdyt?

+#
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+#
+
from unittest import TestCase

import pytest

unit_tests/sources/declarative/validators/test_validate_adheres_to_schema.py (2)

128-130: Test could be more thorough for JSON string inputs

The test for validating JSON strings doesn't make assertions about the actual validation - it just checks that no exception is raised. Consider testing both valid and invalid JSON strings to ensure the validation is working as expected, wdyt?

def test_given_json_string_when_validate_then_succeeds(self):
    schema = {
        "type": "object",
        "properties": {
            "id": {"type": "integer"},
            "name": {"type": "string"},
        },
    }

    validator = ValidateAdheresToSchema(schema=schema)

    validator.validate('{"id": 1, "name": "John Doe"}')
+
+    # Test invalid JSON string
+    with pytest.raises(ValueError) as exc_info:
+        validator.validate('{"id": "not-an-integer", "name": "John Doe"}')
+    
+    assert "integer" in str(exc_info.value)
+
+    # Test malformed JSON string
+    with pytest.raises(ValueError) as exc_info:
+        validator.validate('{bad json}')
+    
+    assert "Invalid JSON" in str(exc_info.value)

---

1-7: Add missing copyright header

The copyright header is missing from this file. All other files in this PR have it, wdyt?

+#
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+#
+
from unittest import TestCase

import jsonschema
import pytest

unit_tests/sources/declarative/validators/test_dpath_validator.py (4)

32-33: Could the assertion be more specific?

The assertion assert strategy.validated_value only checks that the value is truthy. Would it be more precise to assert the exact expected value: assert strategy.validated_value == "test@example.com"? wdyt?

---

91-93: Redundant assertions?

There are two assertions checking the same thing, once with pytest's assert and once with unittest's assertIn. Would it be cleaner to stick with one assertion style for consistency? wdyt?

assert strategy.validate_called
assert strategy.validated_value in ["user1@example.com", "user2@example.com"]
-self.assertIn(strategy.validated_value, ["user1@example.com", "user2@example.com"])

Or if you prefer unittest style:

assert strategy.validate_called
-assert strategy.validated_value in ["user1@example.com", "user2@example.com"]
self.assertIn(strategy.validated_value, ["user1@example.com", "user2@example.com"])

---

23-94: Consider adding more wildcard test cases

The tests are comprehensive, but would it be valuable to add a test with multiple wildcards in the path, like ["users", "*", "addresses", "*", "city"]? This would verify that nested wildcards work correctly. Additionally, the current wildcard test only verifies that the strategy was called with one of the expected values, but not that it was called for all matching paths. wdyt?

---

23-94: Consider adding docstrings to test methods

The tests are well-structured and have descriptive names, but would adding docstrings to explain the purpose and expectations of each test help future developers understand the test cases better? wdyt?

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

airbyte_cdk/sources/declarative/validators/validate_adheres_to_schema.py (1)

36-40: Good error handling with descriptive error messages.

The exception handling here is clean and the error message includes the original validation error, which will be helpful for debugging.

One small thing - have you considered adding a check for empty schemas? An empty schema will validate anything, which might not be the intended behavior in some cases. wdyt?

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

unit_tests/sources/declarative/transformations/config_transformations/test_remove_fields.py (1)

1-101: ⚠️ Potential issue

Fix import file mismatch error

The pipeline is failing with an import file mismatch error, which suggests there might be cached .pyc files or a naming conflict with another test file.

#!/bin/bash
# List all test files with similar names to identify potential conflicts
find unit_tests -name "test_remove_fields*.py"

# Check for __pycache__ directories that might contain conflicting .pyc files
find unit_tests -name "__pycache__" -type d

Have you considered cleaning the pycache directories before running the tests? Alternatively, you could rename this test file to something more unique, like test_remove_fields_transformation.py, wdyt?

🧰 Tools

🪛 GitHub Actions: PyTest Matrix

[error] 1-1: Import file mismatch error: imported module 'test_remove_fields' points to a different file path than the test file collected. Hint: remove pycache / .pyc files or use unique basenames for test modules.

unit_tests/sources/declarative/transformations/config_transformations/test_add_fields.py (1)

1-98: ⚠️ Potential issue

Fix import file mismatch error

Similar to the other test file, there's a pipeline failure due to an import file mismatch error.

#!/bin/bash
# List all test files with similar names to identify potential conflicts
find unit_tests -name "test_add_fields*.py"

# Check for __pycache__ directories that might contain conflicting .pyc files
find unit_tests -name "__pycache__" -type d

Could we rename this test file to something more unique like test_add_fields_transformation.py to avoid conflicts? Or maybe cleaning the pycache directories before running tests would help, wdyt?

🧰 Tools

🪛 GitHub Actions: Pytest (Fast)

[error] 1-1: Import file mismatch error: imported module 'test_add_fields' points to a different file path than the test file to collect. Hint: remove pycache / .pyc files and/or use a unique basename for your test file modules.

🪛 GitHub Actions: PyTest Matrix

[error] 1-1: Import file mismatch error: imported module 'test_add_fields' points to a different file path than the test file collected. Hint: remove pycache / .pyc files or use unique basenames for test modules.

airbyte_cdk/sources/declarative/transformations/config_transformations/remove_fields.py (1)

47-49: Consider validating field_pointers

Have you considered adding validation to ensure that field_pointers is not empty? This could prevent unexpected behavior if an empty list is provided.

Maybe something like this would be helpful, wdyt?

def __post_init__(self) -> None:
+    if not self.field_pointers:
+        logger.warning("RemoveFields instantiated with empty field_pointers list. No fields will be removed.")
    self._filter_interpolator = InterpolatedBoolean(condition=self.condition, parameters={})

airbyte_cdk/sources/declarative/transformations/config_transformations/add_fields.py (4)

1-4: License header looks good.

The copyright year is set to 2025, which is in the future. Is this intentional or should it be the current year instead? wdyt?

---

67-72: Class attributes look well defined.

Good use of field() for internal state management. Would it make sense to add documentation for the condition parameter since it's an important part of the functionality? wdyt?

---

73-103: Initialization logic effectively validates and processes input.

The validation of field paths ensures that empty paths are rejected, which is good defensive programming. One potential improvement: would it be valuable to add validation for the path format beyond just checking length? For instance, ensuring the path elements are all strings or valid dictionary keys?

---

104-121: Transform implementation looks solid but could use a safeguard.

The transform method correctly implements the required behavior, but there's no protection against overwriting existing fields. Would it be helpful to add an optional parameter to control whether overwriting existing fields is allowed? This could prevent accidental data loss in some scenarios.

Also, have you considered adding error handling around the dpath.new call to provide more descriptive error messages if something goes wrong? This would make debugging easier for users.

@dataclass
class AddFields(ConfigTransformation):
    # ... existing attributes ...
+   allow_overwrite: bool = True  # Controls whether existing fields can be overwritten
    
    # ... rest of the class ...
    
    def transform(
        self,
        config: MutableMapping[str, Any],
    ) -> None:
        # ... existing code ...
        
        for parsed_field in self._parsed_fields:
            valid_types = (parsed_field.value_type,) if parsed_field.value_type else None
            value = parsed_field.value.eval(config, valid_types=valid_types)
            is_empty_condition = not self.condition
            if is_empty_condition or self._filter_interpolator.eval(
                config, value=value, path=parsed_field.path
            ):
+               # Check if field exists and respect allow_overwrite setting
+               try:
+                   if not self.allow_overwrite and dpath.util.search(config, parsed_field.path, yielded=True):
+                       continue  # Skip this field as it exists and overwriting is not allowed
                    dpath.new(config, parsed_field.path, value)
+               except Exception as e:
+                   raise ValueError(f"Failed to add field at path {parsed_field.path}: {str(e)}")

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

airbyte_cdk/sources/declarative/transformations/config_transformations/remap_field.py (1)

20-21: Consider supporting dynamic values through InterpolatedMapping?

The static mapping might be limiting for scenarios where we want to interpolate values dynamically. For example, when remapping a field to a value that incorporates parts of the existing config like {{ config.url }}/v1/etc. A previous reviewer noted this might be useful for cases like the Amazon Seller Partner connector, wdyt?

airbyte_cdk/sources/declarative/transformations/config_transformations/__init__.py (1)

5-9: Concise and well-structured exports!

The import structure and __all__ declaration look good, providing a clean public API for the config transformations package.

Have you considered also exporting the base ConfigTransformation class for users who might want to create their own custom transformations? This would make the inheritance hierarchy more discoverable, wdyt?

unit_tests/sources/declarative/transformations/config_transformations/test_config_remap_field.py (1)

68-82: Nice test for interpolated paths

The test effectively demonstrates the dynamic path interpolation capability, which is a powerful feature. Have you considered adding a test case for when the interpolated value doesn't exist or evaluates to something unexpected (like a non-string)? This would ensure robust error handling, wdyt?

unit_tests/sources/declarative/transformations/config_transformations/test_config_remove_fields.py (1)

1-99: Consider adding tests for array element removal

The tests are comprehensive for object field removal, but I noticed that according to the implementation in remove_fields.py, the transformation also supports removing elements from arrays by setting them to None. Would it be valuable to add a test case that verifies this behavior specifically, wdyt?

For example:

def test_given_array_element_field_pointer_element_is_set_to_none():
    transformation = RemoveFields(
        field_pointers=[["array", 1]]  # Remove second element
    )
    config = {"array": ["keep", "remove", "also_keep"]}
    
    transformation.transform(config)
    
    assert config["array"] == ["keep", None, "also_keep"]

airbyte_cdk/sources/declarative/transformations/config_transformations/remap_field.py (2)

23-34: Simplified initialization logic possible?

The post-init logic seems to be doing redundant work. You first create a new field_path list with all elements converted to InterpolatedString, then loop through again to check and convert any remaining strings. Perhaps this could be simplified to a single loop?

- self._field_path = [
-     InterpolatedString.create(path, parameters={}) for path in self.field_path
- ]
- for path_index in range(len(self.field_path)):
-     if isinstance(self.field_path[path_index], str):
-         self._field_path[path_index] = InterpolatedString.create(
-             self.field_path[path_index], parameters={}
-         )
+ self._field_path = []
+ for path in self.field_path:
+     if isinstance(path, str):
+         self._field_path.append(InterpolatedString.create(path, parameters={}))
+     else:
+         self._field_path.append(path)

---

48-60: Consider adding debug or error logging

The transform method silently returns in several cases where the path doesn't exist or intermediate values aren't mappings. Adding debug logging might help with troubleshooting when the transformation doesn't behave as expected. Also, you might want to provide a return value indicating if a transformation was actually applied?

airbyte_cdk/sources/declarative/transformations/config_transformations/remove_fields.py (1)

62-66: Consider adding debug logging

The error handling catches PathNotFound but doesn't log any information. For debugging purposes, it might be helpful to add debug logs when a path is not found. This could help identify potential configuration issues without causing runtime failures.

try:
    dpath.delete(config, pointer)
except dpath.exceptions.PathNotFound:
+   # Using logging here instead of print for proper integration with the framework's logging system
+   import logging
+   logging.debug(f"Path {pointer} not found in config during removal")
    pass

airbyte_cdk/sources/declarative/transformations/config_transformations/add_fields.py (3)

18-25: Well-structured field definition dataclass

The frozen dataclass design is a good choice for immutability of field definitions. One small suggestion - would it make sense to add a docstring explaining what value_type is for and how it's used? It's not immediately obvious from the context.

---

73-102: Post-initialization logic could be simplified

The logic for processing field values is a bit verbose with nested if-else statements. Consider refactoring to reduce nesting:

def __post_init__(self) -> None:
    self._filter_interpolator = InterpolatedBoolean(condition=self.condition, parameters={})

    for add_field in self.fields:
        if len(add_field.path) < 1:
            raise ValueError(
                f"Expected a non-zero-length path for the AddFields transformation {add_field}"
            )

-       if not isinstance(add_field.value, InterpolatedString):
-           if not isinstance(add_field.value, str):
-               raise ValueError(
-                   f"Expected a string value for the AddFields transformation: {add_field}"
-               )
-           else:
-               self._parsed_fields.append(
-                   ParsedAddFieldDefinition(
-                       add_field.path,
-                       InterpolatedString.create(add_field.value, parameters={}),
-                       value_type=add_field.value_type,
-                   )
-               )
-       else:
-           self._parsed_fields.append(
-               ParsedAddFieldDefinition(
-                   add_field.path,
-                   add_field.value,
-                   value_type=add_field.value_type,
-               )
-           )
+       # Handle the value based on its type
+       if isinstance(add_field.value, InterpolatedString):
+           value = add_field.value
+       elif isinstance(add_field.value, str):
+           value = InterpolatedString.create(add_field.value, parameters={})
+       else:
+           raise ValueError(
+               f"Expected a string value for the AddFields transformation: {add_field}"
+           )
+           
+       self._parsed_fields.append(
+           ParsedAddFieldDefinition(
+               add_field.path,
+               value,
+               value_type=add_field.value_type,
+           )
+       )

---

117-119: Consider field-specific conditions

Currently, the same condition is evaluated for all fields. Would it make sense to support field-specific conditions in the AddedFieldDefinition class? This would provide more granular control over when each field is added.

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

unit_tests/sources/declarative/transformations/config_transformations/test_config_remap_field.py (1)

1-129: 💡 Verification agent

🧩 Analysis chain

Format the file to fix linter errors.

The pipeline failures indicate a Ruff format issue. Running ruff format on this file should resolve the formatting issues.

---

🏁 Script executed:

#!/bin/bash
# Check which formatting issues Ruff would fix
ruff format --diff unit_tests/sources/declarative/transformations/config_transformations/test_config_remap_field.py

Length of output: 2159

---

Format test_config_remap_field.py with ruff to fix trailing commas, blank lines, and line wrapping

To resolve the linter errors, you can run:

ruff format -i unit_tests/sources/declarative/transformations/config_transformations/test_config_remap_field.py

Or manually apply the following changes:

• Add a trailing comma after config=config
• Remove the extra blank line before mapping = {...}
• Wrap the two long RemapField(...).transform(...) calls in parentheses

--- a/unit_tests/sources/declarative/transformations/config_transformations/test_config_remap_field.py
+++ b/unit_tests/sources/declarative/transformations/config_transformations/test_config_remap_field.py
@@ -20,7 +20,7 @@
         remap_transform = RemapField(
             field_path=["authorization", "auth_type"],
             map={"client_credentials": "oauth2", "api_key": "key_auth"},
-            config=config
+            config=config,
         )
@@ -113,7 +113,6 @@
     def test_amazon_seller_partner_marketplace_remap_with_interpolated_mapping(self):
-
         mapping = {
             "endpoint": {
                 "ES": "{{ 'https://sellingpartnerapi' if config.environment == 'production' else 'https://sandbox.sellingpartnerapi' }}-eu.amazon.com",
@@ -121,8 +120,12 @@
         production_config = {"environment": "production", "marketplace": "ES"}
-        RemapField(field_path=["marketplace"], map=mapping["endpoint"], config=sandbox_config).transform(sandbox_config)
-        RemapField(field_path=["marketplace"], map=mapping["endpoint"], config=production_config).transform(production_config)
+        RemapField(
+            field_path=["marketplace"], map=mapping["endpoint"], config=sandbox_config
+        ).transform(sandbox_config)
+        RemapField(
+            field_path=["marketplace"], map=mapping["endpoint"], config=production_config
+        ).transform(production_config)

Wdyt?

🧰 Tools

🪛 GitHub Actions: Linters

[error] 1-1: Ruff format check failed. File would be reformatted. Run 'ruff format' to fix code style issues.

airbyte_cdk/sources/declarative/transformations/config_transformations/add_fields.py (5)

1-3: Check the copyright year.

The copyright notice lists 2025, which is a future date. Should this be updated to the current year?

- # Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+ # Copyright (c) 2023 Airbyte, Inc., all rights reserved.

wdyt?

---

73-103: Consider handling unexpected value types more gracefully.

The validation logic correctly handles InterpolatedString and str types, but what happens with other types? Currently, it would raise a ValueError, but the error message only mentions expecting a string value, which might be confusing.

        if not isinstance(add_field.value, InterpolatedString):
            if not isinstance(add_field.value, str):
                raise ValueError(
-                    f"Expected a string value for the AddFields transformation: {add_field}"
+                    f"Expected a string or InterpolatedString value for the AddFields transformation, but got {type(add_field.value).__name__}: {add_field}"
                )

wdyt?

---

104-120: Document overwriting behavior and consider exception handling.

The transform method looks solid! A few things to consider:

1. dpath.new will overwrite existing values at the specified path. This behavior isn't explicitly documented.
2. There's no exception handling for the dpath.new call, which could fail if there are issues with the path or value.

Consider updating the method docstring to mention the overwrite behavior and adding try-except handling:

    def transform(
        self,
        config: MutableMapping[str, Any],
    ) -> None:
        """
        Transforms a config by adding fields based on the provided field definitions.
+       If a field already exists at the specified path, it will be overwritten.

        :param config: The user-provided configuration to be transformed
        """
        for parsed_field in self._parsed_fields:
            valid_types = (parsed_field.value_type,) if parsed_field.value_type else None
            value = parsed_field.value.eval(config, valid_types=valid_types)
            if not self.condition or self._filter_interpolator.eval(
                config, value=value, path=parsed_field.path
            ):
+               try:
                    dpath.new(config, parsed_field.path, value)
+               except Exception as e:
+                   raise ValueError(f"Failed to add field at path {parsed_field.path}: {str(e)}")

wdyt?

---

67-71: Document the condition parameter and its context variables.

The condition parameter enables conditional field addition, which is powerful! However, it would be helpful to document that value and path are available as variables in the condition evaluation, as seen in the transform method.

Consider enhancing the docstring to clarify this:

    fields: List[AddedFieldDefinition]
+   # A boolean expression that determines whether a field should be added.
+   # The expression can reference the config being transformed, as well as
+   # `value` (the evaluated field value) and `path` (the field path).
    condition: str = ""
    _parsed_fields: List[ParsedAddFieldDefinition] = field(
        init=False, repr=False, default_factory=list
    )

wdyt?

---

36-66: Excellent documentation with helpful YAML examples!

The class documentation is thorough and the YAML examples make it easy to understand how to use this transformation. Consider also adding an example of using the conditional feature since it's a powerful capability.

        # by supplying any valid Jinja template directive or expression
        - path: ["two_times_two"]
          value: "{{ 2 * 2 }}"
+
+       # conditional field addition
+       - path: ["optional_field"]
+         value: "{{ config.source_value }}"
+         condition: "{{ value != 'skip_me' }}"

wdyt?

airbyte_cdk/sources/declarative/transformations/config_transformations/remap_field.py (5)

2-2: Copyright year appears to be set to the future.

The copyright year is set to 2025, but we're currently in 2024. Should we update the year to the current one? wdyt?

- # Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+ # Copyright (c) 2024 Airbyte, Inc., all rights reserved.

---

19-20: Should docstring mention interpolated mapping support?

The docstring describes this as a "transformation that remaps a field's value based on a static map", but I can see from the implementation and tests that it also supports interpolated values in the map. Might be helpful to clarify this in the docstring to make the capability more discoverable. wdyt?

- Transformation that remaps a field's value to another value based on a static map.
+ Transformation that remaps a field's value to another value based on a map.
+ The map values can include interpolated strings that reference config values.

---

29-36: Simplify field_path conversion logic.

The current implementation creates _field_path and then potentially overwrites elements in a separate loop. Perhaps we could simplify this to a single operation? wdyt?

- self._field_path = [
-     InterpolatedString.create(path, parameters={}) for path in self.field_path
- ]
- for path_index in range(len(self.field_path)):
-     if isinstance(self.field_path[path_index], str):
-         self._field_path[path_index] = InterpolatedString.create(
-             self.field_path[path_index], parameters={}
-         )
+ self._field_path = [
+     path if isinstance(path, InterpolatedString) else InterpolatedString.create(path, parameters={})
+     for path in self.field_path
+ ]

---

27-28: Consider using a more specific exception type.

The code raises a generic Exception when the field_path is empty. Using a more specific exception type like ValueError could make error handling more consistent across the codebase. wdyt?

- raise Exception("field_path cannot be empty.")
+ raise ValueError("field_path cannot be empty.")

---

44-47: Docstring should mention early return behavior.

The transform method silently returns without making changes if the field path doesn't exist or if intermediate nodes aren't mappings. This is fine, but should be documented in the docstring for clarity. wdyt?

 Transforms a config by remapping a field value based on the provided map.
 If the original value is found in the map, it's replaced with the mapped value.
 If the value is not in the map, the field remains unchanged.
+If any part of the field path doesn't exist or isn't a mapping, no changes are made.

unit_tests/sources/declarative/transformations/config_transformations/test_config_remap_field.py (1)

94-97: Test could be more specific about exception.

The test verifies that an exception is raised with an empty field path, but doesn't check the exception message. Would it be helpful to be more specific about the expected exception type and message? wdyt?

- with pytest.raises(Exception):
+ with pytest.raises(Exception, match="field_path cannot be empty."):

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro